package com.shujia.source

import org.apache.flink.streaming.api.scala._

object Demo1ListSource {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 基于本地集合的source，有界流
     */

    val ds: DataStream[Int] = env.fromCollection(List(1,2,3,4,5))

    //ds.print()

    /**
      * 基于文件构建source  -->有界流
      */
    val lineDS: DataStream[String] = env.readTextFile("data/words.txt")

    val countDS: DataStream[(String, Int)] = lineDS.flatMap(_.split(","))
        .map((_,1))
        .keyBy(_._1)  //元组中第一个值
        .sum(1)  //下标为1的值

    //countDS.print()

    /**
      * 基于socket构建ds  -->无界流
      */

    val socketDS: DataStream[String] = env.socketTextStream("master",8888)

    socketDS.print()



    env.execute()

  }

}
